package com.ficus.road.maintain.process;

import com.ficus.road.maintain.core.model.CarVideoRecord;
import com.ficus.road.maintain.core.model.GpsLocation;
import com.ficus.road.maintain.core.model.InspectPicture;
import com.ficus.road.maintain.dao.CarVideoRecordDao;
import com.ficus.road.maintain.dao.InspectPictureDao;
import com.ficus.road.maintain.service.DataProcessorManager;
import com.ficus.road.maintain.util.AliyunOSSUtil;
import com.ficus.road.maintain.util.LngLonUtil;
import com.ficus.road.maintain.util.queue.BlockingQueueMonitorDecorator;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.protobuf.ByteString;
import ficus.car_demo.FrameMeta;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.http.util.ByteArrayBuffer;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.sql.Timestamp;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.stream.Collectors;
import java.util.zip.GZIPOutputStream;

import static com.ficus.road.maintain.util.AliyunOSSUtil.intToByteArray;


@Slf4j
public class DataWriterProcessor extends Thread {
    public static final String META = "meta";
    public static final String PIC = "pic";
    static private final int DEFAULT_SIZE = 200;
    public static DateFormat sdf = new SimpleDateFormat("yyyyMMdd");
    private final BlockingQueueMonitorDecorator<FrameMeta.SendMetaRequest> blockingDeque;
    AliyunOSSUtil aliyunOSSUtil;
    Map<String, Long> carFilePathMap;
    CarVideoRecordDao carVideoRecordDao;
    InspectPictureDao inspectPictureDao;
    DataProcessorManager dataProcessorManager;

    public DataWriterProcessor(String name, AliyunOSSUtil aliyunOSSUtil, CarVideoRecordDao carVideoRecordDao, InspectPictureDao inspectPictureDao, DataProcessorManager dataProcessorManager) {
        super(name);
        carFilePathMap = Maps.newConcurrentMap();
        this.aliyunOSSUtil = aliyunOSSUtil;
        this.carVideoRecordDao = carVideoRecordDao;
        this.inspectPictureDao = inspectPictureDao;
        this.dataProcessorManager = dataProcessorManager;
        //用的是ArrayBlockingQueue
        this.blockingDeque = new BlockingQueueMonitorDecorator<>(
                "write_oss_queue",
                new ArrayBlockingQueue<>(DEFAULT_SIZE));
    }

    public static byte[] compress(byte[] bytes) {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        GZIPOutputStream gzip;
        try {
            gzip = new GZIPOutputStream(out);
            gzip.write(bytes);
            gzip.close();
        } catch (IOException e) {
            log.error("gzip compress error.", e);
        }
        return out.toByteArray();
    }

    public void put(FrameMeta.SendMetaRequest sendMetaRequest) throws InterruptedException {

        if (CollectionUtils.isEmpty(sendMetaRequest.getFramesList())) {
            return;
        }

        if (blockingDeque.size() >= DEFAULT_SIZE) {
            log.info("queue is full,poll one");
            blockingDeque.poll();
        }

        blockingDeque.put(sendMetaRequest);
    }

    @Override
    public void run() {
        while (true) {
            try {
                FrameMeta.SendMetaRequest sendMetaRequest = blockingDeque.take();
                List<FrameMeta.FrameResult> frameResultList = sendMetaRequest.getFramesList();
                if (CollectionUtils.isNotEmpty(frameResultList)) {
                    String picturePath = sendMetaRequest.getPicturePath();
                    String metaPath = picturePath.replace(PIC, META);
                    ByteArrayBuffer byteArrayBuffer = new ByteArrayBuffer(1024);
                    for (FrameMeta.FrameResult frame : frameResultList) {
                        // 图片快照, 有病害时会有
                        frame.toBuilder().setPictureForDisease(ByteString.EMPTY);
                        byte[] bytes = frame.toByteArray();
                        byteArrayBuffer.append(intToByteArray(bytes.length), 0, 4);
                        byteArrayBuffer.append(bytes, 0, bytes.length);
                    }
                    aliyunOSSUtil.uploadData(metaPath, byteArrayBuffer.toByteArray());
                    saveCarVideoRecordMetaInfo(sendMetaRequest);
                    DataProcessor dataProcessor = dataProcessorManager.getDataProcessorMap().get(sendMetaRequest.getCarId());
                    for (FrameMeta.FrameResult request : frameResultList) {
                        try {
                            //BaseProcessor的blockingQueue放的是单个的FrameResult
                            //先写到oss，然后再处理病害信息
                            dataProcessor.forcePut(request, picturePath, null, sendMetaRequest.getIsCompressed(), sendMetaRequest.getIsCompressed());
                        } catch (InterruptedException e) {
                            log.error("saveCarVideoRecordMetaInfo error", e);
                        }
                    }
                }
            } catch (Exception e) {
                log.error("write oss file met exception", e);
            }
        }
    }

    private void saveCarVideoRecordMetaInfo(FrameMeta.SendMetaRequest sendMetaRequest) {
        long endTime = sendMetaRequest.getEndTime();
        long startTime = sendMetaRequest.getStartTime();
        String carId = sendMetaRequest.getCarId();
        String picturePath = sendMetaRequest.getPicturePath();
        boolean isCompressed = sendMetaRequest.getIsCompressed();
        String metaPath = picturePath.replace(PIC, META);
        CarVideoRecord carVideoRecord = new CarVideoRecord();
        carVideoRecord.setCarId(carId);
        carVideoRecord.setCreationTime(new Date());
        sdf.setTimeZone(TimeZone.getTimeZone("Asia/Shanghai"));
        String formatDate = sdf.format(new Timestamp(startTime));
        carVideoRecord.setDate(formatDate);
        carVideoRecord.setStartTime(startTime);
        carVideoRecord.setPicturePath(picturePath);
        carVideoRecord.setMetaPath(metaPath);
        carVideoRecord.setCompressed(isCompressed);
        carVideoRecord.setEndTime(endTime);
        carVideoRecordDao.saveOrUpdate(carVideoRecord);
        Map<Long, GpsLocation> gpsLocationMap = sendMetaRequest.getFramesList().stream().collect(Collectors.toMap(FrameMeta.FrameResult::getTimeStamp,
                item -> GpsLocation.builder().longitude(item.getGpsResult().getLongitude()).latitude(item.getGpsResult().getLatitude()).build(), (k1, k2) -> k1));
        List<FrameMeta.InspectPic> inspectPicsList = sendMetaRequest.getInspectPicsList();
        List<InspectPicture> inspectPictureList = new ArrayList<>();
        for (FrameMeta.InspectPic inspectPic : inspectPicsList) {
            InspectPicture inspectPicture = new InspectPicture();
            inspectPicture.setCarId(carId);
            inspectPicture.setTimeStamp(inspectPic.getTimestamp());
            inspectPicture.setFlag(0);
            inspectPicture.setUrl(inspectPic.getUrl());
            inspectPicture.setCreationTime(new Date());
            GpsLocation gpsLocation = gpsLocationMap.getOrDefault(inspectPic.getTimestamp(), new GpsLocation(0, 0));
            if (!LngLonUtil.checkItude(gpsLocation.getLongitude(), gpsLocation.getLatitude())){
                gpsLocation.setLatitude(0);
                gpsLocation.setLongitude(0);
            }
            inspectPicture.setGpsLocation(gpsLocation);

            inspectPicture.setLocation(Lists.newArrayList(gpsLocation.getLongitude(), gpsLocation.getLatitude()));
            inspectPictureList.add(inspectPicture);
        }
        if (CollectionUtils.isNotEmpty(inspectPicsList)) {
            inspectPictureDao.insert(inspectPictureList);
        }

    }
}
